package com.study.flume;

import org.apache.flume.Context;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.source.AbstractSource;

/**
 * 自定义Source
 *
 * @author stephen
 * @date 2019-09-04 16:18
 */
public class MySource extends AbstractSource implements Configurable, PollableSource {

    // 全局前后缀
    private String prefix;
    private String subfix;

    @Override
    public void configure(Context context) {
        prefix = context.getString("prefix");
        subfix = context.getString("subfix", "subfix");
    }

    @Override
    public Status process() throws EventDeliveryException {
        Status status = null;
        try {
            for (int i = 0; i < 100; i++) {
                // 接收数据

                // 构造事件对象
                SimpleEvent event = new SimpleEvent();
                // 给事件设置值
                event.setBody((subfix + "-" + i + "-" + subfix).getBytes());
                // 将事件传给channel
                getChannelProcessor().processEvent(event);

                status = Status.READY;

                Thread.sleep(2000);
            }
        } catch (Exception e) {
            e.printStackTrace();
            status = Status.BACKOFF;
        }
        return status;
    }

    @Override
    public long getBackOffSleepIncrement() {
        return 0;
    }

    @Override
    public long getMaxBackOffSleepInterval() {
        return 0;
    }
}
